package com.jj.net.kcp;

import com.jj.config.Constants;
import com.jj.message.ProtoFormat;
import com.jj.message.battle.BattleMsg;
import com.jj.message.battle.UserMessage;
import com.jj.net.ChannelManager;
import com.jj.net.Context;
import com.jj.net.Message;
import com.jj.service.ConvService;
import io.netty.buffer.ByteBuf;
import org.beykery.jkcp.KcpOnUdp;
import org.beykery.jkcp.KcpServer;
import org.beykery.jkcp.KcpThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

import java.util.HashMap;

/**
 * Copyrights: QidianGroup@JJ
 *
 * @author: LuHan
 * @date: 2018/1/15, 10:41
 */
@Component
@ConfigurationProperties("kcp")
public class KcpChannelManager extends KcpServer implements ChannelManager{
    private static Logger logger = LoggerFactory.getLogger(KcpChannelManager.class);

    private HashMap<Integer, Context> conv2Context = new HashMap<>();

    @Autowired
    private KcpChannelListener kcpChannelListener;

    @Autowired
    private ConvService convService;

    private String host;
    private int port;
    private int workerSize;

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getWorkerSize() {
        return workerSize;
    }

    public void setWorkerSize(int workerSize) {
        this.workerSize = workerSize;
    }

    {
        logger.debug("KCPEndpointManager has been created!");
    }

    public Context getContext(KcpOnUdp kcp){
        synchronized (conv2Context){
            Context context = conv2Context.get(kcp.getKcp().getConv());
            if(context == null){
                context = new KcpContextImpl(kcp);
                conv2Context.put(kcp.getKcp().getConv(), context);
                logger.debug("create new kcpchannel for conv:{}, kcp:{}",
                        kcp.getKcp().getConv(), kcp.getKcp());
                Long uid = convService.getUid(kcp.getKcp().getConv());
                if(uid != null){
                    context.setAttr(Constants.USER_ID_KEY, uid);
                    logger.info("bind kcp conv {} to user {}", kcp.getKcp().getConv(), uid);
                }else{
                    logger.error("conv {} not bind with any uid.", kcp.getKcp().getConv());
                }
            }else{
                KcpContextImpl kcpContext = (KcpContextImpl)context;
                KcpChannel kcpChannel = (KcpChannel) kcpContext.getChannel();
                KcpOnUdp old = kcpChannel.getKcpOnUdp();
                if(old != kcp){
                    logger.info("user connected with different kcp connection. {} -> {}",
                            old.getKcp(), kcp.getKcp());
                    kcpChannel.setKcpOnUdp(kcp);
                }
            }
            return context;
        }
    }

    public Context lookupConv(int conv){
        synchronized (conv2Context){
            return conv2Context.get(conv);
        }
    }

    public Context getContextByUid(Long uid){
        int conv = convService.getConv(uid);
        return lookupConv(conv);
    }

    @Override
    public boolean start() {
        logger.debug("prepare to start kcp userservice");
        setStream(false);
        noDelay(1,20,2,1);
        setMinRto(10);
        wndSize(128, 128);
        setTimeout(60 * 1000);
        setMtu(1400);
        logger.debug("KCPServer is listening at {}, workersize:{}",
                port, workerSize);
        super.start(port, workerSize);
        return true;
    }

    @Override
    public boolean stop() {
        close();
        return false;
    }

//    int idx = 1;

    @Override
    public void handleReceive(ByteBuf bb, KcpOnUdp kcp) {
        Context context = getContext(kcp);

        UserMessage userMsg = UserMessage.parseFrom(bb);
        if(userMsg.getProtocolFormat() != ProtoFormat.Heartbeat)
        {
            kcpChannelListener.onRcv(userMsg, context);
        }else{
            logger.debug("ignore heartbeat {}", userMsg);
        }
    }

    @Override
    public void handleException(Throwable ex, KcpOnUdp kcp) {
        logger.error("exception", ex.getCause());
    }

    @Override
    public void handleClose(KcpOnUdp kcp) {
        logger.debug("kcp {} closed", kcp.getKcp());
        Context context = getContext(kcp);
        kcpChannelListener.onClosed(context);
    }

    public void resetKCP(Long uid){
        Integer conv = convService.lookup(uid);
        if(conv != null){
            for(KcpThread worker : workers){
                worker.remove(conv);
            }
        }else{
            logger.debug("conv not found, ignore reset");
        }
    }
}
